package cn.com.cfae.iras.doc.analyze.config;

import cn.com.cfae.iras.doc.analyze.mq.consumer.DocAnalyzeConsumerPullTaskCallback;
import org.apache.rocketmq.client.consumer.MQPullConsumerScheduleService;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * Created by 佟媛媛 on 18/10/23.
 * RocketMQ相关配置
 */

@Configuration
public class MQConsumerConfig {

    private static Logger logger = LoggerFactory.getLogger(MQConsumerConfig.class);

    @Value("${rocketmq.namesrv.addr}")
    private String namesrvAddr;
    @Value("${rocketmq.group.consumer}")
    private String groupConsumer;
    @Value("${rocketmq.topic.doc_analyze}")
    private String topicDocAnalyze;
    @Value("${rocketmq.doc_analyze.consumer.status}")
    private Boolean docAnalyzeConsumerStatus;


    @Autowired
    private DocAnalyzeConsumerPullTaskCallback analyzeStatusConsumerPullTaskCallback = null;


    @Bean(name = "docAnalyzeConsumer")
    public MQPullConsumerScheduleService getMQPullAnalyzeStatusConsumerScheduleService() {
        MQPullConsumerScheduleService scheduleService = new MQPullConsumerScheduleService(groupConsumer + "_" + topicDocAnalyze);
        scheduleService.getDefaultMQPullConsumer().setNamesrvAddr(namesrvAddr);
        scheduleService.setMessageModel(MessageModel.CLUSTERING);
        scheduleService.registerPullTaskCallback(topicDocAnalyze, analyzeStatusConsumerPullTaskCallback);
        try {
            if (docAnalyzeConsumerStatus) scheduleService.start();
        } catch (MQClientException e) {
            logger.error(e.getMessage(), e);
        }
        return scheduleService;
    }


}
